package com.study.flink.window

import java.text.SimpleDateFormat

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
  * Event Time Window Demo
  *
  * @author stephen
  * @date 2019-07-22 16:35
  */
object ScalaEventTimeWindowDemo {

  def main(args: Array[String]): Unit = {
    // 1 获取执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置为Event Time，默认为Processing Time
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)

    // 2 获取输入数据
    import org.apache.flink.api.scala._
    //a,1563850800000 => 2019-07-23 11:00:00
    //a,1563850803000 => 2019-07-23 11:00:03
    //a,1563850805000 => 2019-07-23 11:00:05
    //a,1563850808000 => 2019-07-23 11:00:08
    //a,1563850812000 => 2019-07-23 11:00:12
    //a,1563850816000 => 2019-07-23 11:00:16
    //a,1563850822000 => 2019-07-23 11:00:22
    //a,1563850815000 => 2019-07-23 11:00:15
    //a,1563850835000 => 2019-07-23 11:00:35
    val dataStream = env.socketTextStream("localhost", 9999)

    // 3 Transformation
    val resultStream = dataStream
      .map(x => {
        val arr = x.split(",")
        // word 时间戳 次数
        (arr(0), arr(1).toLong, 1)
      })
      // 周期性生成水印（AssignerWithPeriodicWatermarks）， 可以设置周期，env.getConfig.setAutoWatermarkInterval(100)
      // 带断点的水印，基于某些事件触发生成水印（AssignerWithPunctuatedWatermarks）
      // 这里的BoundedOutOfOrdernessTimestampExtractor是周期性水印，设置最大容忍乱序时间为10秒
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[(String, Long, Int)](Time.seconds(10)) {
      override def extractTimestamp(element: (String, Long, Int)): Long = {
        element._2
      }
    })
      .keyBy(_._1)
      // 滚动窗口：每10秒计算一次
      //.window(TumblingEventTimeWindows.of(Time.seconds(10)))
      // 滑动窗口：每5秒钟计算最近10秒钟的数据
      .window(SlidingEventTimeWindows.of(Time.seconds(10),Time.seconds(5)))
      .apply(new WindowFunction[(String, Long, Int),String,String,TimeWindow]{
        override def apply(key: String, window: TimeWindow, input: Iterable[(String, Long, Int)], out: Collector[String]): Unit = {
          val sb = new StringBuilder()
          sb.append("======================================\n")
          val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")
          sb.append("窗口开始时间：").append(sdf.format(window.getStart)).append("\n")

          val it = input.iterator
          // 前面没有使用聚合，因此只有每个窗口的第一条数据
          if(it.hasNext){
            val e = it.next()
            sb.append(e._1).append("=>").append(sdf.format(e._2)).append("\n")
          }

          sb.append("窗口结束时间：").append(sdf.format(window.getEnd)).append("\n")
          sb.append("======================================\n")
          out.collect(sb.toString())
        }
      })

    // 4 输出
    resultStream.print().setParallelism(1)

    //======================================
    //窗口开始时间：2019-07-23 10:59:55.000
    //a=>2019-07-23 11:00:00.000
    //窗口结束时间：2019-07-23 11:00:05.000
    //======================================
    //
    //======================================
    //窗口开始时间：2019-07-23 11:00:00.000
    //a=>2019-07-23 11:00:00.000
    //窗口结束时间：2019-07-23 11:00:10.000
    //======================================
    //
    //======================================
    //窗口开始时间：2019-07-23 11:00:05.000
    //a=>2019-07-23 11:00:05.000
    //窗口结束时间：2019-07-23 11:00:15.000
    //======================================
    //
    //======================================
    //窗口开始时间：2019-07-23 11:00:10.000
    //a=>2019-07-23 11:00:12.000
    //窗口结束时间：2019-07-23 11:00:20.000
    //======================================
    //
    //======================================
    //窗口开始时间：2019-07-23 11:00:15.000
    //a=>2019-07-23 11:00:16.000
    //窗口结束时间：2019-07-23 11:00:25.000
    //======================================

    // 5 启动任务
    env.execute("Event Time Window Demo")
  }
}
